package com.whit.bigdata.weblog.pre;

import java.io.IOException;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.HashSet;
import java.util.Set;

import com.whit.bigdata.weblog.mrbean.WebLogBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * 0 处理原始日志，过滤出真实pv请求 转换时间格式 对缺失字段填充默认值 对记录标记valid和invalid
 */

public class WeblogPreProcess extends Configured implements Tool {
//   继承： private Configuration conf;
//   实现：int run(String [] args) throws Exception;


    @Override
    public int run(String[] args) throws Exception {
        //Configuration conf = new Configuration();
        Configuration conf = super.getConf();
        Job job = Job.getInstance(conf);
	 
//        FileInputFormat.addInputPath(job, new Path("C:\\Users\\17438\\Desktop\\大数据课程资料\\t006hive实战\\日志文件数据\\input"));
//        FileOutputFormat.setOutputPath(job, new Path("file:///C:\\Users\\17438\\Desktop\\大数据课程资料\\t006hive实战\\日志文件数据\\weblogPreOut"));
        
        //设置读取工具类 ////如果是打包运行 加上这2句
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        
        //如果是打包运行，可以通过获得args参数，来设置输入和输出路径
        if(args.length != 2) {
        	System.out.println("输入参数错误");
        	return 1;
        }
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //如果是打包运行，执行下面的这句
        job.setJarByClass(WeblogPreProcess.class);
        
        job.setMapperClass(WeblogPreProcessMapper.class);
        //设置map的输出的<k2,v2>的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        
        //设置reduce任务为0，没有ruduce处理
        job.setNumReduceTasks(0);
        boolean res = job.waitForCompletion(true);
        return res ? 0 : 1;
    }

    static class WeblogPreProcessMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        // 用来存储网站url分类数据
        Set<String> pages = new HashSet<String>();
        Text k = new Text();
        NullWritable v = NullWritable.get();

        /**
         * 从外部配置文件中加载网站的有用url分类数据 存储到maptask的内存中，用来对日志数据进行过滤
         */
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
        	//pages列表，凡是请求地址属于下面的，就是有用的数据
            pages.add("/about");
            pages.add("/black-ip-list/");
            pages.add("/cassandra-clustor/");
            pages.add("/finance-rhive-repurchase/");
            pages.add("/hadoop-family-roadmap/");
            pages.add("/hadoop-hive-intro/");
            pages.add("/hadoop-zookeeper-intro/");
            pages.add("/hadoop-mahout-roadmap/");

        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //读入原始文件每一行
        	String line = value.toString();
        	//转换为WebLogBean，如果为null，不处理（说明清洗掉）
            WebLogBean webLogBean = WebLogParser.parser(line);
            if (webLogBean != null) {
                // 过滤js/图片/css等静态资源
            	// 不是属于pages资源访问列表中的数据记录，我们给valid字段打上false值
                WebLogParser.filtStaticResource(webLogBean, pages);
                /* if (!webLogBean.isValid()) return; */

                k.set(webLogBean.toString());
                context.write(k, v);
            }
        }
    }

    public static void main(String[] args) throws Exception {
    	//下面的就是MR的提交运行
        Configuration configuration = new Configuration();
        int run = ToolRunner.run(configuration, new WeblogPreProcess(), args);
        System.exit(run);
    }

}
